package org.indexador.hadoop;

import java.io.*;
import java.util.*;

import org.apache.hadoop.conf.*;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.input.*;
import org.apache.hadoop.mapreduce.lib.output.*;
import org.indexador.util.*;

public class ArtigoWikipediaDriver {
  public static void main(String[] args) throws IOException,
      InterruptedException, ClassNotFoundException {
    long inicio = System.currentTimeMillis();
    Configuration conf = new Configuration();
    //
    Job job = new Job(conf, "Processador Wikipedia");
    //
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(IntWritable.class);
    job.setInputFormatClass(TextInputFormat.class);
    job.setOutputFormatClass(TextOutputFormat.class);
    //
    Formatter formatter = new Formatter();
    String outpath = System.getProperty("user.home") + "/dados/hadoop-out/"
        + formatter.format("%1$tm%1$td%1$tH%1$tM%1$tS", new Date());
    //    FileInputFormat.setInputPaths(job, new Path(Constantes.INDICE_WIKIPEDIA_STAX));
    FileInputFormat.setInputPaths(job, new Path(
        Constantes.ARQUIVO_ACORDAOS_HDFS));
    FileOutputFormat.setOutputPath(job, new Path(outpath));
    job.setMapperClass(ArtigoWikipediaMapper.class);
    job.setReducerClass(ArtigoWikipediaReducer.class);
    System.out.println(job.waitForCompletion(true));
    long fim = System.currentTimeMillis();
    long tempoEmMinutos = ((fim - inicio) / 1000 / 60);
    String unidadeDeTempo = " minuto(s)";
    if (tempoEmMinutos == 0) {
      unidadeDeTempo = " segundo(s)";
      tempoEmMinutos = ((fim - inicio) / 1000);
    }
    System.out.println("Tempo de processamento: " + tempoEmMinutos
        + unidadeDeTempo);
  }
}
